package com.jc.demo.zk;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Random;

import static org.apache.zookeeper.ZooDefs.Ids.OPEN_ACL_UNSAFE;

public class AsyncMaster implements Watcher {
    ZooKeeper zk;
    String hostPort;

    String serverId = Long.toString(new Random().nextLong());
    public boolean isLeader = false;

    AsyncMaster(String hostPort) {
        this.hostPort = hostPort;
    }

    void startZK() throws IOException {
        zk = new ZooKeeper(hostPort, 15000, this);
    }

    public void process(WatchedEvent e) {
        System.out.println(e);
    }

    public void stopZK() throws InterruptedException {
        zk.close();
    }


    void runFormaster() {
        zk.create("/master", serverId.getBytes(), OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, new AsyncMasterCreateCallback(), null);

    }

    void checkMaster() {
        zk.getData("/master", false, new AsyncDataCallBack(), null);
    }

    public static void main(String args[])
            throws Exception {
        AsyncMaster m = new AsyncMaster("vm1.jevoncode.com:2181,vm2.jevoncode.com:2181,vm3.jevoncode.com:2181");
        m.startZK();

        m.runFormaster();


        Thread.sleep(60000);
        m.stopZK();
    }


    class AsyncMasterCreateCallback implements AsyncCallback.StringCallback {

        @Override
        public void processResult(int rc, String path, Object ctx, String name) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    checkMaster();
                    return;
                case OK:
                    isLeader = true;
                    break;
                default:
                    isLeader = false;
            }
            System.out.println("I'm " + (isLeader ? "" : "not") + " the leader");
        }
    }

    class AsyncDataCallBack implements AsyncCallback.DataCallback {

        @Override
        public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    checkMaster();
                case NONODE:
                    runFormaster();
                    return;
            }
        }
    }

}